package com.chis.jmdatatimer.main;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitPolicy;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.core.io.ClassPathResource;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.TreeSet;

/**
 * 
 * @Description 
 * @Company zwx
 * @author wlj
 * @version 2019年01月02日
 */
public class HbaseSplit {
    private static Logger log = LogManager.getLogger(HbaseSplit.class);

    private static String splitkey=null;
    private static String compress=null;
    private static String splitprefix=null;

    public static void main(String[] args) throws IOException {
        System.out.println(args[0]);
        System.out.println(args[1]);
        System.out.println(args[2]);
        System.out.println(args[3]);

        Properties properties = new Properties();
        properties.load(new ClassPathResource("config.properties").getInputStream());
//        compress = properties.getProperty("hbase.compress");
        compress = args[3];
//        String hbasetype = properties.getProperty("hbase.splittype");
        String hbasetype = args[2];
        if(hbasetype.equals("0")){
            splitkey = "3202,3203,3204,3205,3206,3207,3208,3209,3210,3211,3212,3213";
            splitprefix = "4";
        } else {
            splitkey = "1,2,3,4,5,6,7,8,9,A,B,C,D,E,F";
            splitprefix = "1";
        }


        HbaseSplit test = new HbaseSplit();
//        String table = properties.getProperty("hbase.table");
        String table = args[0];
        List<String> list = new ArrayList<>();
//        list.add(properties.getProperty("hbase.family"));
        list.add(args[1]);
        String s = test.createTableBySplitKeys(table,list);
        System.out.println(s);
        log.error(s);
    }

    private byte[][] getSplitKeys() {
        String[] keys = splitkey.split(",");
        byte[][] splitKeys = new byte[keys.length][];
        TreeSet<byte[]> rows = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
        for (int i = 0; i < keys.length; i++) {
            rows.add(Bytes.toBytes(keys[i]));
        }
        Iterator<byte[]> rowKeyIter = rows.iterator();
        int i=0;
        while (rowKeyIter.hasNext()) {
            byte[] tempRow = rowKeyIter.next();
            rowKeyIter.remove();
            splitKeys[i] = tempRow;
            i++;
        }
        return splitKeys;
    }

    /**
     * 创建预分区hbase表
     * @param tableName 表名
     * @param columnFamily 列簇
     * @return
     */
    @SuppressWarnings("resource")
    public String createTableBySplitKeys(String tableName, List<String> columnFamily) {
        try {
            if (StringUtils.isBlank(tableName) || columnFamily == null || columnFamily.size() < 0) {
                log.error("===入参不能为空===");
            }
            Configuration configuration = HBaseConfiguration.create();
            HBaseAdmin admin = new HBaseAdmin(configuration);
            if (admin.tableExists(tableName)) {
                return "表已存在";
            } else {
                HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
                tableDescriptor.setValue(HTableDescriptor.SPLIT_POLICY, KeyPrefixRegionSplitPolicy.class.getName());
                tableDescriptor.setValue("KeyPrefixRegionSplitPolicy.prefix_length", splitprefix);
                HColumnDescriptor columnDesc = null;
                for (String cf : columnFamily) {
                    columnDesc = new HColumnDescriptor(Bytes.toBytes(cf));
                    if(compress.endsWith("1")){
                        columnDesc.setCompactionCompressionType(Compression.Algorithm.SNAPPY);
                    }
                    tableDescriptor.addFamily(columnDesc);
                }
                byte[][] splitKeys = getSplitKeys();
                admin.createTable(tableDescriptor,splitKeys);//指定splitkeys
                admin.close();
                log.info("===建表： " + tableName
                        + " 列簇:" + columnFamily.toString()
                        + "===");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e);
            return "预分区建表失败";
        }
        return "预分区建表成功";
    }


    private void createTable() throws IOException {
        //HashChoreWoker worker = new HashChoreWoker(1000000, 10);
        //byte[][] splitKeys = worker.calcSplitKeys();

        byte[][] splitKeys = getSplitKeys();

        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();
        TableName tableName = TableName.valueOf("TD_DATA");

        if (admin.tableExists(tableName)) {
            try {
                admin.disableTable(tableName);
            } catch (Exception e) {
                e.printStackTrace();
            }
            admin.deleteTable(tableName);
        }

        HTableDescriptor tableDesc = new HTableDescriptor(tableName);
        tableDesc.setValue(HTableDescriptor.SPLIT_POLICY, KeyPrefixRegionSplitPolicy.class.getName());
        tableDesc.setValue("KeyPrefixRegionSplitPolicy.prefix_length", "1");
        HColumnDescriptor columnDesc = new HColumnDescriptor(Bytes.toBytes("info"));
        columnDesc.setMaxVersions(1);
        //columnDesc.setCompactionCompressionType(Compression.Algorithm.SNAPPY);
        tableDesc.addFamily(columnDesc);
        admin.createTable(tableDesc, splitKeys);
        admin.close();
    }


}
